from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode, ReduceFunction

# 1、创建flink执行环境
env = StreamExecutionEnvironment.get_execution_environment()
# 修改并行度
env.set_parallelism(1)

# 2、读取数据,得到DataStream，相当于RDD  （有界流）
lines_ds = env.read_text_file("G:\LanZhiPeiXun\Flink\data\words.txt")

# 统计单词的数量
# 一行转换成多行
words_ds = lines_ds.flat_map(lambda line: line.split(","))
# 转换成kv格式
kv_ds = words_ds.map(lambda word: (word, 1))
# 按照单词分组
key_by_ds = kv_ds.key_by(lambda kv: kv[0])


# reduce: keyby之后对相同的key进行聚合计算
# reduce计算是有状态计算
# 状态:之前的计算结果
# count_ds = key_by_ds.reduce(lambda x, y: (x[0], x[1] + y[1])).print()


# 类和对象的方式
class CountReduceFunction(ReduceFunction):

    # x: 之前的结果，y: 当前数据
    def reduce(self, x, y):
        print(f"x: {x}, y: {y}")
        return x[0], x[1] + y[1]


key_by_ds.reduce(CountReduceFunction()).print()

env.execute()
